package com.jscloud.bigdata.flink.flinksql.functions.udf;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;

/**
 * # FlinkSQL自定义函数一进一出实战UDF
 * 通过FlinkSQL当中的自定义函数ScalarFunction来实现json解析，通过传入json的key，获取到json对应的value字段值
 *
 * json数据格式内容如下
 * {"date_time":"2022-10-04 08:01:48","email":"kyzqcd0686@vjikq.tng","id":0,"name":"郑剃"}
 * {"date_time":"2022-10-04 08:06:31","email":"bvkqwbmgwi@lh80q.4ln","id":1,"name":"闾丘喜造"}
 * {"date_time":"2022-10-04 08:04:39","email":"axvcbj7vbo@ecyi1.4gw","id":2,"name":"王曙介"}
 * {"date_time":"2022-10-04 08:00:19","email":"ew1qu5sunz@caxtg.vtn","id":3,"name":"赖溯姆"}
 *
 * 定义json解析函数
 * ##如果自定义函数，必须要继承ScalarFunction类##
 */
public class JsonParseFunction extends ScalarFunction {

        @Override
        public void open(FunctionContext context) throws Exception {
                //初始化方法，如果有一些初始化工作，都写在open方法里
        }

        /**
         *  通过定义 eval 方法来实现自定义函数的逻辑
         * @param jsonLine
         * @param key
         * @return
         */
        public String eval(String jsonLine, String key){
                JSONObject jsonObject = JSONObject.parseObject(jsonLine);
                if(jsonObject.containsKey(key)) {
                        return   jsonObject.getString(key);
                } else {
                        return "";
                }
        }

        @Override
        public void close() throws Exception {
                // 结束时候调用的方法，如果有一些清理类的工作都写在 close 方法里
        }
}